//异步工作器类
#ifndef _M_LOOPER_H__
#define _M_LOOPER_H__

#include"buffer.hpp"
#include<mutex>
#include<condition_variable>
#include<memory>
#include<atomic>
#include<thread>
#include<functional>

namespace bitlog{
    class Asynclooper{
        public:
        using ptr=shared_ptr<Asynclooper>;
        using Functor=function<void(Buffer& buffer)>;
        Asynclooper(const Functor& callback):_running(true),_callback(callback),_thread(thread(&Asynclooper::worker_loop,this))
        {}
        ~Asynclooper() {stop();}
        void stop(){
            _running=false;
            _pop_cond.notify_all();
            _thread.join();
        }
        void push(const string& msg)
        {
            {
                unique_lock<mutex> lock(_mutex);
                _push_cond.wait(lock,[&]{
                    return _task_push.writeAblesize()>msg.size();
                });
                _task_push.push(msg.c_str(),msg.size());
            }

            //std::cout << "push 成功" << std::endl;
          //  std::cout << "_task_push.empty(): " << _task_push.empty() << std::endl;
            _pop_cond.notify_one();
          //  sleep(3);
        }
        protected:
        void worker_loop(){
            while(1){

                {
                    unique_lock<mutex> lock(_mutex);
                    //std::cout << pthread_self() << std::endl;
                    if(!_running && _task_push.empty()) 
                    {
                        std::cout << "-----------------------" << std::endl;
                        cout << "消费者线程退出" << std::endl;
                        break;
                    }
                    _pop_cond.wait(lock,[&]{
                        return (!_task_push.empty() || !_running);
                    });
                    _task_pop.swap(_task_push);
                }
                _push_cond.notify_one();
                _callback(_task_pop);
                _task_pop.reset();
            }
        }
        private:
        Functor _callback;
        private:
        mutex _mutex;
        condition_variable _push_cond;
        condition_variable _pop_cond;

        atomic<bool>  _running;
        thread _thread;
       
        Buffer _task_push;
        Buffer _task_pop;

    };
}

#endif